package com.dlc.apk.core

import android.content.Context
import com.dlc.apk.App
import com.dlc.apk.api.ApiConstants
import com.dlc.apk.api.entity.*
import com.dlc.apk.base.BaseApp
import com.dlc.apk.core.ShipmentTaskHandler
import com.dlc.apk.dao.ObjectBox
import com.dlc.apk.dao.entity.ShipmentOrder
import com.dlc.apk.dao.entity.ShipmentOrder_
import com.dlc.apk.service.log.ILogService
import com.dlc.apk.service.log.impl.LogService
import com.dlc.apk.utils.*
import com.dlc.hardwareprotocol.serialport.command.recv.RecvBA
import com.jeremyliao.liveeventbus.LiveEventBus
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.disposables.Disposable
import io.reactivex.schedulers.Schedulers
import org.eclipse.paho.android.service.MqttAndroidClient
import org.eclipse.paho.client.mqttv3.*
import java.io.File
import java.util.concurrent.ArrayBlockingQueue
import java.util.concurrent.ThreadPoolExecutor
import java.util.concurrent.TimeUnit

class MQTTClient internal constructor(private var builder: Builder? = null) {
    private var mqttClient: MqttAndroidClient? = null
    private var mqttClientId: String? = ""

    object InstanceHolder {
        val instance: MQTTClient = MQTTClient(null)
    }

    fun connect(context: Context, brokerURI: String, userName: String, password: String, mqttCallback: IMQTTCallback) {
        val clientId = "${DeviceUtil.packageName}_${System.currentTimeMillis()}"
        mqttClientId = clientId
        LogPlus.e("###MQTT CLIENT 开始连接...### clientId->${clientId}")
        mqttClient = MqttAndroidClient(context, brokerURI, clientId)
        mqttClient!!.setCallback(object : MqttCallbackExtended {
            override fun messageArrived(topic: String?, message: MqttMessage?) {
                LogPlus.d("Receive message: ${message.toString()} from topic: $topic")
                mqttCallback.messageArrived(topic, message)
            }

            override fun connectionLost(cause: Throwable?) {
                LogPlus.d("Connection lost ${cause.toString()}")
                mqttCallback.connectionLost(cause)
                //重连
                BaseApp.instance.uiHandler?.postDelayed({
                    connect(context, brokerURI, userName, password, mqttCallback)
                }, 1000 * 3)
            }

            override fun deliveryComplete(token: IMqttDeliveryToken?) {
                mqttCallback.deliveryComplete(token)
            }

            override fun connectComplete(reconnect: Boolean, serverURI: String?) {
                LogPlus.e("###MQTT connectComplete! reconnect -> $reconnect, serverURI->$serverURI")
            }
        })
        val options = MqttConnectOptions()
        options.isCleanSession = true
        options.connectionTimeout = 30
        options.keepAliveInterval = 20
        options.isAutomaticReconnect = false /*true*/ //手动重连
        options.setWill("nm_will", "{}".toByteArray(), 0, false)
        options.userName = userName
        options.password = password.toCharArray()
        try {
            mqttClient!!.connect(options, null, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    LogPlus.d("### MQTT Connection success")
                    mqttCallback.onSuccess(asyncActionToken)
                }

                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    LogPlus.d("Connection failure")
                    mqttCallback.onFailure(asyncActionToken, exception)
                    //重连
                    BaseApp.instance.uiHandler?.postDelayed({
                        connect(context, brokerURI, userName, password, mqttCallback)
                    }, 1000 * 3)
                }
            })
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }

    fun subscribe(topic: String, qos: Int = 1) {
        try {
            mqttClient!!.subscribe(topic, qos, null, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    LogPlus.d("Subscribed to $topic")
                }

                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    LogPlus.d("Failed to subscribe $topic")
                }
            })
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }

    fun subscribe(topics: Array<String>, QoSs: IntArray) {
        mqttClient?.subscribe(topics, QoSs)
    }

    fun unsubscribe(topic: String) {
        try {
            mqttClient?.unsubscribe(topic, null, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    LogPlus.d("Unsubscribed to $topic")
                }

                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    LogPlus.d("Failed to unsubscribe $topic")
                }
            })
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }

    fun publish(topic: String, msg: String, qos: Int = 1, retained: Boolean = false, actionListener: IMqttActionListener? = null) {
        try {
            val message = MqttMessage()
            message.payload = msg.toByteArray()
            message.qos = qos
            message.isRetained = retained
            mqttClient?.publish(topic, message, null, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    LogPlus.d("$msg published to $topic")
                    actionListener?.onSuccess(asyncActionToken)
                }

                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    LogPlus.d("Failed to publish $msg to $topic, 异常堆栈信息->${LogPlus.getStackTraceInfo(exception)}")
                    actionListener?.onFailure(asyncActionToken, exception)
                }
            })
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }

    fun disconnect() {
        try {
            mqttClient?.disconnect(null, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    LogPlus.d("Disconnected")
                }

                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    LogPlus.d("Failed to disconnect")
                }
            })
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }

    fun isConnected(): Boolean {
        return mqttClient?.isConnected ?: false
    }

    interface IMQTTCallback {
        fun connectionLost(cause: Throwable?)

        @Throws(Exception::class)
        fun messageArrived(topic: String?, message: MqttMessage?)

        fun deliveryComplete(token: IMqttDeliveryToken?)

        fun onSuccess(asyncActionToken: IMqttToken?)

        fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?)
    }

    class Builder constructor() {
        internal lateinit var brokerUrl: String
        internal lateinit var clientId: String

        internal lateinit var context: Context

        internal var cleanSession = true
        internal var automaticReconnect = true

        internal var keepAliveInterval = 60

        internal var mqttCallback: MqttCallback? = null

        fun brokerUrl(brokerUrl: String) = apply {
            this.brokerUrl = brokerUrl
        }

        fun clientId(clientId: String) = apply {
            this.clientId = clientId
        }

        fun context(context: Context) = apply {
            this.context = context.applicationContext
        }

        fun cleanSession(cleanSession: Boolean) = apply {
            this.cleanSession = true
        }

        fun automaticReconnect(automaticReconnect: Boolean) = apply {
            this.automaticReconnect = automaticReconnect
        }

        fun keepAliveInterval(keepAliveInterval: Int) = apply {
            this.keepAliveInterval = keepAliveInterval
        }

        fun mqttCallback(mqttCallback: MqttCallback?) = apply {
            this.mqttCallback = mqttCallback
        }

        fun build(): MQTTClient = MQTTClient(this)
    }

    companion object {
        private val threadPoolExecutor = ThreadPoolExecutor(20, 100, 1000, TimeUnit.SECONDS, ArrayBlockingQueue(100))
        private var heartDataSendingDisposable: Disposable? = null
        val logService: ILogService = LogService()

        fun getInstance(): MQTTClient {
            return InstanceHolder.instance
        }

        //后台发布的topic
        val serverPublishTopic = ApiConstants.serverPublishTopic
        val serverPublishTopic2 = ApiConstants.serverPublishTopic2

        //后台订阅的topic
        val serverSubscribeTopic = ApiConstants.serverSubscribeTopic

        //初始化
        fun init() {
            val serverUri = ApiConstants.brokerUri
            val userName = ApiConstants.brokerUserName
            val password = ApiConstants.brokerPassword
            getInstance().connect(BaseApp.instance, serverUri, userName, password, object : MQTTClient.IMQTTCallback {
                override fun connectionLost(cause: Throwable?) {
                    LogPlus.e("###MQTT连接丢失###")
                }

                override fun messageArrived(topic: String?, message: MqttMessage?) {
                    LogPlus.e("###MQTT消息到达###")
                    if (ApiConstants.uploadLogMQTTTopic == topic) {
                        LogPlus.e("###MQTT上传日志...")
                        uploadLog()
                        return
                    }
                    if (message == null) {
                        return
                    }
                    LogPlus.e("###MQTT 消息 topic->$topic, payload->${String(message.payload)},msgId->${message.id}")
                    if (topic == serverSubscribeTopic) {
                        LogPlus.e("###可能来自APK发布的消息### topic->$topic, payload->${String(message.payload)},msgId->${message.id}")
                        return
                    }
                    val baseMQTTMsg = GsonUtil.instance.parseJsonStrToObj(String(message.payload), BaseMQTTMsg::class.java)
                    if (baseMQTTMsg == null) {
                        LogPlus.e("###MQTT 消息数据格式异常###")
                        return
                    }
                    when (baseMQTTMsg.api_name) {
                        "restart" -> {
                            LogPlus.e("###MQTT restart###")
                            LiveEventBus.get(ApiConstants.EVENT_MQTT_RESTART).post("")
                            BaseApp.instance.restartApp(1000)
                            DeviceUtil.restart()
                        }
                        "update_adv" -> {
                            LogPlus.e("###MQTT update_adv###")
                            LiveEventBus.get(ApiConstants.EVENT_MQTT_UPDATE_ADVERT).post("")
                        }
                        "updateApk" -> {//app版本/嵌入式版本 更新
                            LogPlus.e("###MQTT updateApk###")
                            val updateVersionMQTTMsg = GsonUtil.instance.parseJsonStrToObj(String(message.payload), UpdateVersionMQTTMsg::class.java)
                            LiveEventBus.get(ApiConstants.EVENT_MQTT_UPDATE_VERSION).post(updateVersionMQTTMsg)
                            CacheHelper.saveApkVersion(updateVersionMQTTMsg)
                        }
                    }
                }

                override fun deliveryComplete(token: IMqttDeliveryToken?) {
                    //LogPlus.e("###MQTT消息已投递完成###")
                }

                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    LogPlus.e("###MQTT连接成功###")
                    /////////////////////////订阅/////////////////////////
                    getInstance().subscribe(arrayOf(serverPublishTopic, serverPublishTopic2, serverSubscribeTopic, ApiConstants.uploadLogMQTTTopic), intArrayOf(0, 0, 0, 0))
                    /////////////////////////订阅/////////////////////////
                    /*开始发布心跳数据到后台*/
                    heartDataSendingDisposable?.dispose()
                    sendHeartDataToServer()
                }

                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    LogPlus.e("###MQTT连接失败###")
                }
            })
        }

        //释放
        fun release() {
            LogPlus.e("###释放MQTT client###")
            heartDataSendingDisposable?.dispose()
            getInstance().disconnect()
        }

        var workStatusData: RecvBA? = null

        private val heartData: HeartData = HeartData()

        fun sendHeartDataToServer() {
            heartDataSendingDisposable?.dispose()
            heartDataSendingDisposable = io.reactivex.Observable.interval(5, 30, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.computation())
                .doOnSubscribe { } //执行计时任务前
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe {
                    publishHeart(workStatusData)
                }
        }

        private const val MIN_HEART_DATA_DELAY = 1000 * 3
        private var lastPublishHeartDataTime: Long = 0
        private val heartDataMap = HashMap<String, Any>()


        fun publishHeart(data: RecvBA?) {
            if (System.currentTimeMillis() - lastPublishHeartDataTime < MIN_HEART_DATA_DELAY) {
                LogPlus.e("###小于最低上报间隔,暂不上报###")
                return
            }
            val topic = serverSubscribeTopic
            heartDataMap["api_name"] = "health"
            heartDataMap["macno"] = DeviceUtil.androidId
            if (data != null) {
                heartData.flowNo = data.flowNo
                heartData.shipmentState = data.shipmentState
                heartData.currentStep = data.currentStep
                heartData.errCode = data.errCode
                heartData.th1ShipmentResult = data.th1ShipmentResult
                heartData.th2ShipmentResult = data.th2ShipmentResult
                heartData.th3ShipmentResult = data.th3ShipmentResult
                heartData.th4ShipmentResult = data.th4ShipmentResult
                heartData.th5ShipmentResult = data.th5ShipmentResult
                heartData.outState = data.outState
                heartData.inState = data.inState
                heartData.auxiliaryInPulseCount1 = data.auxiliaryInPulseCount1
                heartData.auxiliaryInPulseCount2 = data.auxiliaryInPulseCount2
                heartData.temperature = data.temperature
                heartData.deviceType = data.deviceType
            }
            val adList: ArrayList<AdStatisticsData> = ArrayList()//广告数据
            CacheHelper.getAdResponseData()?.forEachIndexed { index, adData ->
                adList.add(AdStatisticsData(adData.id.toString(), CacheHelper.getAdPlayTimes(adData.id.toString())))
            }
            heartData.adData = adList
            heartDataMap["data"] = heartData
            heartDataMap["clientId"] = getInstance().mqttClientId ?: ""
            heartDataMap["clientTime"] = DateUtil.nowToString()
            heartDataMap["appVersion"] = DeviceUtil.packageVersion ?: ""

            getInstance().publish(topic, GsonUtil.instance.parseObjToJsonStr(heartDataMap), 0, false, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    lastPublishHeartDataTime = System.currentTimeMillis()
                }

                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                }
            })
        }

        fun uploadLog() {
            //上传本地日志到服务器
            if (LogPlus.todayLogFile != null) {
                Thread {
                    val targetLogFile = File("${LogPlus.todayLogFile?.absolutePath}${DeviceUtil.androidId}_.log")
                    LogPlus.todayLogFile?.copyTo(targetLogFile, true, 2048)
                    logService.uploadLog(targetLogFile)
                }.start()
            }
        }

        class HeartData {
            var flowNo = 0
            var shipmentState = 0
            var currentStep = 0
            var errCode = 0
            var th1ShipmentResult = 0
            var th2ShipmentResult = 0
            var th3ShipmentResult = 0
            var th4ShipmentResult = 0
            var th5ShipmentResult = 0
            var outState = 0
            var inState = 0
            var auxiliaryInPulseCount1 = 0
            var auxiliaryInPulseCount2 = 0
            var temperature: String? = null
            var deviceType = 0

            var adData: List<AdStatisticsData>? = null
        }

        data class AdStatisticsData(var id: String?, var nums: Int = 0)
    }
}